package com.aizuda.snailjob.core;

import cn.hutool.core.util.HashUtil;
import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import cn.hutool.db.Page;
import cn.hutool.db.PageResult;
import cn.hutool.db.sql.Direction;
import cn.hutool.db.sql.Order;
import com.aizuda.snailjob.core.Collector.GroupConfigCollector;
import com.aizuda.snailjob.core.Collector.JobCollector;
import com.aizuda.snailjob.core.Collector.NamespaceCollector;
import com.aizuda.snailjob.enums.ExecutorBlockStrategyEnum;
import com.aizuda.snailjob.enums.ExecutorRouteStrategyEnum;
import com.aizuda.snailjob.enums.ScheduleTypeEnum;
import com.aizuda.snailjob.util.CronUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import javax.sql.DataSource;
import java.security.SecureRandom;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public final class DataMigrationTool {

    public static final Collector COLLECTOR = new Collector();

    public static void migrationNamespace(DataSource targetDS, String name, String namespaceId) {

        Entity insertEntity = Entity.create("sj_namespace")
            .set("name", name)
            .set("unique_id", namespaceId)
            .set("description", "自动工具迁移生成");

        NamespaceCollector namespaceCollector = COLLECTOR.getNamespaceCollector();

        try {
            Db.use(targetDS).insertOrUpdate(insertEntity, "unique_id");
            namespaceCollector.getSuccess().incrementAndGet();
        } catch (SQLException e) {
            log.error("命名空间数据迁移失败. namespaceId:[{}] name:[{}]", namespaceId, name, e);
            namespaceCollector.getError().incrementAndGet();
        }

    }

    public static Map<Long, String> migrationGroup(DataSource sourceDS, DataSource targetDS, String namespaceId) throws SQLException {
        GroupConfigCollector groupConfigCollector = COLLECTOR.getGroupConfigCollector();

        Map<Long, String> map = new HashMap<>();
        Page page = new Page();
        page.addOrder(new Order("id", Direction.ASC));
        List<Entity> list = Db.use(sourceDS).findAll(Entity.create("xxl_job_group"));
        groupConfigCollector.setTotal(list.size());
        AtomicLong success = groupConfigCollector.getSuccess();
        AtomicLong error = groupConfigCollector.getError();
        list.forEach(entity -> {
            // 为后续的job迁移做准备
            map.put(entity.getLong("id"), entity.getStr("app_name"));
            Entity insertEntity = Entity.create("sj_group_config")
                .set("namespace_id", namespaceId)
                .set("group_name", entity.getStr("app_name"))
                .set("description", entity.getStr("title"))
                .set("token", generateToken(32))
                .set("group_partition", 0)
                .set("group_status", 1)
                .set("init_scene", 1)
                .set("version", 1);

            try {

                Db.use(targetDS).insertOrUpdate(insertEntity, "namespace_id", "group_name");
                success.incrementAndGet();
            } catch (SQLException e) {
                log.error("Group数据迁移失败. jobName:[{}]", insertEntity.getStr("job_name"), e);
                error.incrementAndGet();
            }
        });

        return map;
    }

    public static void migrationJob(DataSource sourceDS, DataSource targetDS, Map<Long, String> groupMap,
        String namespaceId, int totalBucket)
        throws SQLException {

        JobCollector jobCollector = COLLECTOR.getJobCollector();
        Page page = new Page();
        page.setPageSize(100);
        page.addOrder(new Order("id", Direction.ASC));
        PageResult<Entity> pageResult = Db.use(sourceDS)
            .page(Entity.create("xxl_job_info"), page);
        int totalPage = pageResult.getTotalPage();
        jobCollector.setTotal(pageResult.getTotal());

        for (int i = 1; i <= totalPage; i++) {
            doMigrationJob(targetDS, groupMap, namespaceId, totalBucket, pageResult, jobCollector);
            page.setPageNumber(i);
            pageResult = Db.use(sourceDS).page(Entity.create("xxl_job_info"), page);
            if (pageResult.isEmpty()) {
                break;
            }
        }

    }

    private static void doMigrationJob(final DataSource targetDS, final Map<Long, String> groupMap, final String namespaceId,
        final int totalBucket, final PageResult<Entity> pageResult, final JobCollector jobCollector) {
        pageResult.forEach(entity -> {

            String jobName = entity.getStr("job_desc") + "-" + entity.get("id");
            try {
                // 幂等处理
                List<Entity> entities = Db.use(targetDS).find(Entity.create("sj_job").set("job_name", jobName));
                if (!CollectionUtils.isEmpty(entities)) {
                    return;
                }
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }

            //  2. 固定时间 3.CRON 表达式 99.工作流
            Integer triggerType = null;
            String scheduleType = entity.getStr("schedule_type");
            if (ScheduleTypeEnum.NONE.name().equals(scheduleType)) {
                // 此类型不做任何处理
                return;
            } else if (ScheduleTypeEnum.CRON.name().equals(scheduleType)) {
                triggerType = 3;
            } else if (ScheduleTypeEnum.FIX_RATE.name().equals(scheduleType)) {
                triggerType = 2;
            }

            ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.valueOf(
                entity.getStr("executor_block_strategy"));
            ExecutorRouteStrategyEnum routeStrategy = ExecutorRouteStrategyEnum.valueOf(
                entity.getStr("executor_route_strategy"));

            Entity insertEntity = Entity.create("sj_job")
                .set("namespace_id", namespaceId)
                .set("group_name", groupMap.get(entity.getLong("job_group")))
                .set("job_name", entity.getStr("job_desc") + "-" + entity.get("id"))
                .set("args_str", entity.getStr("executor_param"))
                .set("next_trigger_at", entity.getLong("trigger_next_time"))
                .set("job_status", entity.getInt("trigger_status"))
                .set("task_type", routeStrategy.name().equals("SHARDING_BROADCAST") ? 2 : 1)
                .set("route_key", routeStrategy.getSjRoute())
                .set("executor_type", 1)
                .set("executor_info", entity.getStr("executor_handler"))
                .set("trigger_type", triggerType)
                .set("trigger_interval", entity.getStr("schedule_conf"))
                .set("block_strategy", blockStrategy.getSjBlockStrategy())
                .set("executor_timeout", entity.getInt("executor_timeout") == 0 ? 60 : entity.getStr("executor_timeout"))
                .set("max_retry_times", entity.getStr("executor_fail_retry_count"))
                .set("retry_interval", 1)
                .set("parallel_num", 1)
                .set("resident", isResident(triggerType, entity.getStr("schedule_conf")))
                .set("description", entity.getStr("job_desc"))
                .set("create_dt", entity.getDate("add_time"))
                .set("update_dt", entity.getDate("update_time"))
                .set("bucket_index", HashUtil.bkdrHash(entity.getStr("job_desc")) % totalBucket);

            try {
                Db.use(targetDS).insert(insertEntity);
                jobCollector.getSuccess().incrementAndGet();
            } catch (SQLException e) {
                log.error("定时任务数据迁移失败. jobName:[{}]", insertEntity.getStr("job_name"), e);
                jobCollector.getError().incrementAndGet();
            }
        });
    }

    private static Integer isResident(Integer triggerType, String triggerInterval) {

        if (triggerType == 2) {
            if (Integer.parseInt(triggerInterval) < 10) {
                return 0;
            }
        } else if (triggerType == 3) {
            if (CronUtils.getExecuteInterval(triggerInterval) < 10 * 1000) {
                return 1;
            }
        } else {
            throw new IllegalArgumentException("未知触发类型");
        }

        return 0;
    }

    /**
     * 生成 token
     */
    public static String generateToken(int length) {
        String chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
        SecureRandom random = new SecureRandom();
        StringBuilder token = new StringBuilder("SJ_");

        for (int i = 0; i < length; i++) {
            int randomNumber = random.nextInt(chars.length());
            token.append(chars.charAt(randomNumber));
        }

        return token.toString();
    }
}

